package com.xdxc.service.consumer;

import com.xdxc.mapper.TempUserAnswerMapper;
import com.xdxc.model.entity.TempUserAnswer;
import com.xdxc.service.serviceinterface.GradingResultService;
import com.xdxc.service.serviceinterface.ProcessedBatchService;
import com.xdxc.service.serviceinterface.TempUserAnswerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.ibatis.cursor.Cursor;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.stream.Collectors;


@Component
@RocketMQMessageListener(
        topic = "EXAM_ANSWER_TOPIC",
        consumerGroup = "scoring-group",
        consumeMode = ConsumeMode.ORDERLY // 保证同一examId的顺序处理
)
@Slf4j
public class AnswerConsumer implements RocketMQListener<List<TempUserAnswer>> {

    @Autowired
    private TempUserAnswerService tempUserAnswerService;

    @Autowired
    private GradingResultService gradingResultService;
    @Autowired
    private ProcessedBatchService processedBatchService;

    @Override
    @Transactional(rollbackFor = Exception.class)  // 确保事务覆盖整个处理流程
    public void onMessage(List<TempUserAnswer> batch) {
        if (CollectionUtils.isEmpty(batch)) {
            return;
        }

        // 0. 生成唯一批次ID（examId + 答案列表哈希）
        Long examId = batch.get(0).getExamId();
        String batchId = generateBatchId(examId, batch);

        // 1. 幂等检查
        if (processedBatchService.isProcessed(batchId)) {
            log.info("[幂等拦截] 批次已处理，跳过: examId={}, batchId={}", examId, batchId);
            return;
        }

        try {
            // 2. 存储原始答案（批量插入）
//            tempUserAnswerService.saveBatch(batch);

            // 3. 分批评分（每100条一批）
            int batchSize = 100;
            for (int i = 0; i < batch.size(); i += batchSize) {
                List<TempUserAnswer> scoringBatch = batch.subList(i, Math.min(i + batchSize, batch.size()));
                gradingResultService.objectGrade(scoringBatch, examId);
            }
            // 4. 标记批次完成（与上述操作同一事务）
            processedBatchService.markProcessed(batchId, examId);
            log.info("批次处理成功: examId={}, batchId={}, size={}", examId, batchId, batch.size());
        } catch (Exception e) {
            log.error("批次处理失败: examId={}, batchId={}", examId, batchId, e);
            throw new RuntimeException("消息消费失败", e);  // 触发事务回滚
        }
    }

    /**
     * 生成唯一批次ID（examId + 答案ID列表的MD5）
     */
    private String generateBatchId(Long examId, List<TempUserAnswer> batch) {
        String answerIds = batch.stream()
                .map(answer -> answer.getId().toString())
                .sorted()  // 排序确保相同答案列表生成相同哈希
                .collect(Collectors.joining("|"));
        String hash = DigestUtils.md5Hex(answerIds);
        return examId + "_" + hash;
    }
}